package com.zyx.flinkdemo.sql.catalog;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Schema;
import org.apache.flink.table.api.SqlDialect;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.catalog.Catalog;
import org.apache.flink.table.catalog.CatalogBaseTable;
import org.apache.flink.table.catalog.ObjectPath;
import org.apache.flink.table.catalog.exceptions.TableNotExistException;
import org.apache.flink.table.catalog.hive.HiveCatalog;
import org.apache.hadoop.hive.metastore.api.FieldSchema;
import org.apache.hadoop.hive.metastore.api.Table;

import java.util.List;
import java.util.Map;
import java.util.function.Consumer;

/**
 * @author zyx
 * @since 2021/5/31 21:29
 * desc: HiveCatalog案例
 * @see <a href="https://ci.apache.org/projects/flink/flink-docs-release-1.13/zh/docs/connectors/table/hive/hive_read_write/">Flink的Hive读写</a>
 * @see <a href="https://ci.apache.org/projects/flink/flink-docs-release-1.13/zh/docs/connectors/table/filesystem/#streaming-sink">Streaming Sink</a>
 */
public class HiveCatalogDemo {
    public static void main(String[] args) throws Exception {
        // 获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        EnvironmentSettings settings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build();
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env, settings);

        // 创建HiveCatalog
        HiveCatalog hiveCatalog = new HiveCatalog("myHive", "db_test", "input/hive");

        // 注册HiveCatalog
        tableEnv.registerCatalog("myHive", hiveCatalog);

        // 使用HiveCatalog
        tableEnv.useCatalog("myHive");

        // 查询当前库下的表
        /*List<String> tables = hiveCatalog.listTables("db_test");
        for (String table : tables) {
            System.out.println(table);
        }*/

        // 查询表格信息
        /*tableEnv.getCatalog("myHive").ifPresent(catalog -> {
            // 查询表格是否存在
            ObjectPath hiveTbop = new ObjectPath("db_test", "hive_ticker");
            System.out.println("db_test.hive_ticker是否存在?" + catalog.tableExists(hiveTbop));
            try {
                CatalogBaseTable table = catalog.getTable(hiveTbop);
                // 查询列信息
                for (Schema.UnresolvedColumn column : table.getUnresolvedSchema().getColumns()) {
                    System.out.println(column.getName());
                }
            } catch (TableNotExistException e) {
                e.printStackTrace();
            }
        });*/

        // 查看分区信息
        /*ObjectPath hiveTbop = new ObjectPath("db_test", "hive_ticker");
        Table hiveTable = hiveCatalog.getHiveTable(hiveTbop);
        for (FieldSchema partitionKey : hiveTable.getPartitionKeys()) {
            System.out.println(partitionKey);
        }*/



        /*
            Hive建表语句:
            CREATE TABLE db_test.hive_ticker (
                `securityCode` INT,
                `price` DOUBLE,
                `sendTime` BIGINT
            ) PARTITIONED BY (dt STRING, hr STRING)
            ROW FORMAT DELIMITED FIELDS TERMINATED BY '\t'
            LOCATION '/warehouse/db_test/hive_ticker';
         */

        // 执行查询,查询Hive中已经存在的表数据
        // tableEnv.executeSql("select * from hive_ticker").print();

        /*
            {
              "aggregateQuantity": 6000,
              "filler": 0,
              "msgSize": 36,
              "msgType": 52,
              "price": 4460.0,
              "securityCode": 806,
              "sendTime": 1628058978325,
              "seqNum": 45607125,
              "tickerId": 408,
              "tradeTime": 1628058978000000000,
              "trdCancelFlag": "N",
              "trdType": 0
            }
         */

        // 插入数据,插入数据到Hive中
        tableEnv.executeSql("INSERT INTO db_test.hive_ticker PARTITION (dt='2021-08-04', hr='14', mi='07') SELECT 2321, 68352.0, 1628057158509");

    }
}
